package org.raftkv.raft.core.log;

import org.raftkv.commons.Tuple;
import org.raftkv.raft.core.Raft;
import org.raftkv.raft.core.storage.Storage;
import org.raftkv.raft.core.storage.exceptions.CompactedException;
import org.raftkv.raft.core.storage.exceptions.UnavailableException;
import org.raftkv.raft.exceptions.RaftException;
import org.raftkv.raft.pb.Entry;
import org.raftkv.raft.pb.Snapshot;
import org.raftkv.raft.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * raft日志的相关操作
 */
public class RaftLog {

    private Storage storage;

    /**
     * 保存还没有持久化的数据
     */
    private Unstable unstable;

    /**
     * 保存当前提交的日志数据索引
     */
    private long committed;

    /**
     * 保存已经应用到状态机的数据的最高索引
     */
    private long applied;

    private static final Logger LOGGER = LoggerFactory.getLogger(RaftLog.class);

    private long maxNextEntsSize;

    private RaftLog() {

    }

    private RaftLog(Storage storage, long maxNextEntsSize) {
        this.storage = storage;
        this.maxNextEntsSize = maxNextEntsSize;
        this.unstable = new Unstable();
    }

    public static RaftLog newLog(Storage storage) {
        return newLogWithSize(storage, Raft.NO_LIMIT);
    }

    public static RaftLog newEmptyLog() {
        return new RaftLog();
    }

    public static RaftLog newLogWithSize(Storage storage, long maxNextEntsSize) {
        if (storage == null) {
            throw new RaftException("storage must not be null");
        }
        RaftLog log = new RaftLog(storage, maxNextEntsSize);
        long firstIndex = storage.firstIndex();
        long lastIndex = storage.lastIndex();

        // offset从持久化之后的最后一个index的下一个开始
        log.unstable.setOffset(lastIndex + 1);
        // committed和applied从持久化的第一个index的前一个开始
        log.committed = firstIndex - 1;
        log.applied = firstIndex - 1;

        return log;
    }

    @Override
    public String toString() {
        return String.format("committed=%d, applied=%d, unstable.offset=%d, len(unstable.Entries)=%d",
                committed, applied, unstable.getOffset(), unstable.getEntries().size());
    }

    /**
     * 尝试添加一组日志，如果不能添加则返回(0,false)，否则返回(新的日志的索引,true)
     *
     * @param index     从哪里开始的日志条目
     * @param logTerm   这一组日志对应的term
     * @param committed leader上的committed索引
     * @param entries   需要提交的一组日志，因此这组数据的最大索引为index+len(ents)
     */
    public Tuple<Long, Boolean> maybeAppend(long index, long logTerm, long committed, List<Entry> entries) {
        // 首先需要保证传入的index和logTerm能匹配的上才能走入这里，否则直接返回false
        if (matchTerm(index, logTerm)) {
            if (entries == null) {
                entries = List.of();
            }

            // 首先得到传入数据的最后一条索引
            long lastnewi = index + entries.size();
            // 查找传入的数据从哪里开始找不到对应的Term了
            long ci = findConflict(entries);
            if (ci == 0) {
                ;
            } else if (ci <= this.committed) {
                // 找到的数据索引小于committed，都说明传入的数据是错误的
                String errMsg = String.format("entry %d conflict with committed entry [committed(%d)]", ci, this.committed);
                LOGGER.error(errMsg);
                throw new RaftException(errMsg);
            } else {
                // ci > 0的情况下来到这里
                long offset = index + 1;
                // 从查找到的数据索引开始，将这之后的数据放入到unstable存储中
                append(entries.subList((int) (ci - offset), entries.size()));
            }
            // 选择committed和lastnewi中的最小者进行commit
            commitTo(Math.min(committed, lastnewi));
            return new Tuple<>(lastnewi, true);
        }
        return new Tuple<>(0L, false);
    }

    /**
     * @param ents 要添加的日志条目
     * @return 最后一条日志的索引
     */
    public long append(List<Entry> ents) {
        if (ents.size() == 0) {
            return lastIndex();
        }
        // 如果索引小于committed，则说明该数据是非法的
        long after = ents.get(0).getIndex() - 1;
        if (after < committed) {
            String errMsg = String.format("after(%d) is out of range [committed(%d)]", after, committed);
            LOGGER.error(errMsg);
            throw new RaftException(errMsg);
        }
        // 放入unstable存储中
        unstable.truncateAndAppend(ents);
        return lastIndex();
    }

    /**
     * 返回第一个在entry数组中，index中的term与当前存储的数据不同的索引
     * 如果没有冲突数据，而且当前存在的日志条目包含所有传入的日志条目，返回0；
     * 如果没有冲突数据，而且传入的日志条目有新的数据，则返回新日志条目的第一条索引
     * 一个日志条目在其索引值对应的term与当前相同索引的term不相同时认为是有冲突的数据。
     *
     * @return index中的term与当前存储的数据不同的索引
     */
    public long findConflict(List<Entry> entries) {
        for (Entry ne : entries) {
            // 找到第一个任期号不匹配的，即当前在raftLog存储的该索引数据的任期号，不是ent数据的任期号
            if (!matchTerm(ne.getIndex(), ne.getTerm())) {
                if (ne.getIndex() <= lastIndex()) {
                    // 如果不匹配任期号的索引数据，小于当前最后一条日志索引，就打印错误日志
                    String msg = String.format("found conflict at index %d [existing term: %d, conflicting term: %d]",
                            ne.getIndex(), zeroTermOnErrCompacted(ne.getIndex()), ne.getTerm());
                    LOGGER.info(msg);
                }
                return ne.getIndex();
            }
        }
        return 0;
    }

    public long findConflictByTerm(long index, long term) {
        long li = lastIndex();
        if (index > li) {
            String msg = String.format("index(%d) is out of range [0, lastIndex(%d)] in findConflictByTerm",
                    index, li);
            LOGGER.info(msg);
            return index;
        }

        while (true) {
            long logTerm = term(index);
            if (logTerm <= term) {
                break;
            }
            index--;
        }
        return index;
    }

    public List<Entry> unstableEntries() {
        if (unstable.getEntries().size() == 0) {
            return null;
        }
        return unstable.getEntries();
    }

    public List<Entry> nextEnts() {
        long off = Math.max(applied + 1, firstIndex());
        if (committed + 1 > off) {
            List<Entry> ents;
            try {
                ents = slice(off, committed + 1, maxNextEntsSize);
            } catch (Exception e) {
                String errMsg = String.format("unexpected error when getting unapplied entries (%s)", e);
                LOGGER.error(errMsg);
                throw new RaftException(errMsg);
            }
            return ents;
        }
        return null;
    }

    public boolean hasNextEnts() {
        long off = Math.max(applied + 1, firstIndex());
        return committed + 1 > off;
    }

    public boolean hasPendingSnapshot() {
        return unstable.getSnapshot() != null && !Snapshot.isEmptySnap(unstable.getSnapshot());
    }

    /**
     * 返回快照数据
     *
     * @return snapshot
     */
    public Snapshot snapshot() {
        if (unstable.getSnapshot() != null) {
            // 如果没有保存的数据有快照，就返回
            return unstable.getSnapshot();
        }
        return storage.snapshot();
    }

    /**
     * @return 第一条日志的索引
     */
    public long firstIndex() {
        // 首先尝试在未持久化数据中看有没有快照数据
        Tuple<Long, Boolean> t = unstable.maybeFirstIndex();
        Long i = t.getFirst();
        Boolean ok = t.getSecond();
        if (ok) {
            return i;
        }
        // 否则才返回持久化数据的firstIndex
        return storage.firstIndex();
    }

    /**
     * @return 最后一条日志的索引
     */
    public long lastIndex() {
        Tuple<Long, Boolean> t = unstable.maybeLastIndex();
        Long i = t.getFirst();
        Boolean ok = t.getSecond();
        if (ok) {
            return i;
        }
        return storage.lastIndex();
    }

    public void commitTo(long toCommit) {
        // 首先需要判断，commit索引绝不能变小
        if (committed < toCommit) {
            // 传入的值如果比lastIndex大则是非法的
            if (lastIndex() < toCommit) {
                String errMsg = String.format("tocommit(%d) is out of range [lastIndex(%d)]. Was the raft log corrupted, truncated, or lost?",
                        toCommit, lastIndex());
                LOGGER.error(errMsg);
                throw new RaftException(errMsg);
            }
            committed = toCommit;
        }
    }

    public void appliedTo(long i) {
        if (i == 0) {
            return;
        }
        if (committed < i || i < this.applied) {
            String errMsg = String.format("applied(%d) is out of range [prevApplied(%d), committed(%d)]",
                    i, this.applied, committed);
            LOGGER.error(errMsg);
            throw new RaftException(errMsg);
        }
        this.applied = i;
    }

    /**
     * 传入数据索引，该索引表示在这个索引之前的数据应用层都进行了持久化，修改unstable的数据
     *
     * @param i index
     * @param t term
     */
    public void stableTo(long i, long t) {
        unstable.stableTo(i, t);
    }

    public void stableSnapTo(long i) {
        unstable.stableSnapTo(i);
    }

    /**
     * @return 最后一条日志的term
     */
    public long lastTerm() {
        try {
            return term(lastIndex());
        } catch (Throwable e) {
            LOGGER.error("unexpected error when getting the last term");
            throw new RaftException("unexpected error when getting the last term", e);
        }
    }

    public long term(long i) {
        // 合法的范围是 [index of dummy entry, last index]
        long dummyIndex = firstIndex() - 1;

        // 先判断范围是否正确
        if (i < dummyIndex || i > lastIndex()) {
            return 0;
        }

        Tuple<Long, Boolean> temp = unstable.maybeTerm(i);
        Long t = temp.getFirst();
        Boolean ok = temp.getSecond();
        if (ok) {
            return t;
        }
        try {
            return storage.term(i);
        } catch (CompactedException | UnavailableException e) {
            return 0;
        }
    }

    /**
     * 获取从i开始的entries返回，大小不超过maxsize
     *
     * @param index   开始的index
     * @param maxSize 最大大小
     */
    public List<Entry> entries(int index, long maxSize) {
        if (index > lastIndex()) {
            return null;
        }
        return slice(index, lastIndex() + 1, maxSize);
    }

    /**
     * @return raftLog中的所有entry
     */
    public List<Entry> allEntries() {
        try {
            return entries((int) firstIndex(), Raft.NO_LIMIT);
        } catch (CompactedException e) {
            // try again if there was a racing compaction
            return allEntries();
        }
    }

    /**
     * 判断是否比当前节点的日志更新：1）term是否更大 2）term相同的情况下，索引是否更大
     */
    public boolean isUpToDate(long lasti, long term) {
        return term > lastTerm() || (term == lastTerm() && lasti >= lastIndex());
    }

    public boolean matchTerm(long i, long term) {
        long t = term(i);
        return t == term;
    }

    public boolean maybeCommit(long maxIndex, long term) {
        // 只有在传入的index大于当前commit索引，以及index对应的term与传入的term匹配时，才使用这些数据进行commit
        if (maxIndex > committed && zeroTermOnErrCompacted(maxIndex) == term) {
            commitTo(maxIndex);
            return true;
        }
        return false;
    }

    public void restore(Snapshot s) {
        LOGGER.info("log [{}] starts to restore snapshot [index: {}, term: {}]",
                this, s.getMetadata().getIndex(), s.getMetadata().getTerm());
        committed = s.getMetadata().getIndex();
        unstable.restore(s);
    }

    /**
     * 返回[lo,hi-1]之间的数据，这些数据的大小总和不超过maxSize
     *
     * @param lo      上界
     * @param hi      下界 -1
     * @param maxSize 最大总数据大小
     * @return [lo, hi-1]之间的Entry
     */
    public List<Entry> slice(long lo, long hi, long maxSize) {
        mustCheckOutOfBounds(lo, hi);
        if (lo == hi) {
            return null;
        }
        List<Entry> ents = new ArrayList<>();
        // lo 小于unstable的offset，说明前半部分在持久化的storage中
        if (lo < unstable.getOffset()) {
            // 注意传入storage.Entries的hi参数取hi和unstable offset的较小值
            List<Entry> storeEntries;
            try {
                storeEntries = storage.entries(lo, Math.min(hi, unstable.getOffset()), maxSize);
            } catch (CompactedException e) {
                throw e;
            } catch (UnavailableException e) {
                String errMsg = String.format("entries[%d:%d) is unavailable from storage",
                        lo, Math.min(hi, unstable.getOffset()));
                LOGGER.error(errMsg);
                throw new RaftException(errMsg);
            } catch (Throwable e) {
                throw new RaftException(e);
            }
            if (storeEntries.size() < Math.min(hi, unstable.getOffset()) - lo) {
                return storeEntries;
            }
            ents = storeEntries;
        }
        if (hi > unstable.getOffset()) {
            // hi大于unstable offset，说明后半部分在unstable中取得
            List<Entry> unstableEntries = unstable.slice(Math.max(lo, unstable.getOffset()), hi);
            if (ents.size() > 0) {
                ArrayList<Entry> combined = new ArrayList<>(ents);
                combined.addAll(unstableEntries);
                ents = combined;
            } else {
                ents = unstableEntries;
            }
        }
        return Utils.limitSize(ents, maxSize);
    }

    public void mustCheckOutOfBounds(long lo, long hi) {
        if (lo > hi) {
            String errMsg = String.format("invalid slice %d > %d", lo, hi);
            LOGGER.error(errMsg);
            throw new RaftException(errMsg);
        }
        long fi = firstIndex();
        if (lo < fi) {
            throw new CompactedException();
        }
        long len = lastIndex() + 1 - fi;
        if (hi > fi + len) {
            String errMsg = String.format("slice[%d,%d) out of bound [%d,%d]", lo, hi, fi, lastIndex());
            LOGGER.error(errMsg);
            throw new RaftException(errMsg);
        }
    }

    /**
     * 如果term 函数抛出 CompactedException 异常，则返回0
     *
     * @return index对应的term
     */
    public long zeroTermOnErrCompacted(long index) {
        try {
            return term(index);
        } catch (CompactedException ex) {
            return 0;
        } catch (Exception e) {
            LOGGER.error("unexpected error ({})", e, e);
            return 0;
        }
    }

    public Storage getStorage() {
        return storage;
    }

    public void setStorage(Storage storage) {
        this.storage = storage;
    }

    public Unstable getUnstable() {
        return unstable;
    }

    public void setUnstable(Unstable unstable) {
        this.unstable = unstable;
    }

    public long getCommitted() {
        return committed;
    }

    public void setCommitted(long committed) {
        this.committed = committed;
    }

    public long getApplied() {
        return applied;
    }

    public void setApplied(long applied) {
        this.applied = applied;
    }

    public long getMaxNextEntsSize() {
        return maxNextEntsSize;
    }

    public void setMaxNextEntsSize(long maxNextEntsSize) {
        this.maxNextEntsSize = maxNextEntsSize;
    }

}
